package com.mjf.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mjf.bean.OrderWide;
import com.mjf.bean.PaymentInfo;
import com.mjf.bean.PaymentWide;
import com.mjf.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;

/**
 * 订单宽表与支付信息表生成支付宽表
 *
 * 数据流向：web/app -> nginx -> springboot -> MySQL -> flinkApp -> kafka(ods) -> flinkApp -> kafka(dwd)/HBase(dim)
 *         -> flinkApp(redis) -> Kafka(dwm) -> flinkApp -> Kafka(dwm)
 * 程序：mockDb(包含web/nginx/springboot) -> MySQL -> FlinkCDC(包含flinkApp/kafka(ods)) -> BaseDbApp(包含flinkApp/kafka(dwd)/HBase(dim))
 *      -> OrderWideApp(包含flinkApp/Kafka(dwm)) -> PaymentWideApp(包含flinkApp/Kafka(dwm))
 * 环境：hdfs zookeeper kafka hbase phoenix(bin/sqlline.py) redis
 */
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {

        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

/*
        // 测试时关闭
        // 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9000/gmall-flink/checkpoint"));
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);

        // 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10));
*/

        // 2.读取 Kafka 主题的数据创建流，并转换为JavaBean对象，提取时间戳生成watermark
        String groupId = "PaymentWideApp";
        String orderWideTopic = "dwm_order_wide";
        String paymentInfoTopic = "dwd_payment_info";
        String sinkTopic = "dwm_payment_wide";

        SingleOutputStreamOperator<OrderWide> orderWideDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderWideTopic, groupId))
                .map(line -> JSON.parseObject(line, OrderWide.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderWide>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderWide>() {
                                    @Override
                                    public long extractTimestamp(OrderWide element, long recordTimestamp) {
                                        try {
                                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                            return sdf.parse(element.getCreate_time()).getTime();
                                        } catch (ParseException e) {
                                            e.printStackTrace();
                                            return recordTimestamp; // 发生解析异常时用进入系统的时间
                                        }
                                    }
                                })
                );

//        orderWideDS.print("orderWideDS======>");

        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = env.addSource(MyKafkaUtil.getKafkaConsumer(paymentInfoTopic, groupId))
                .map(line -> JSON.parseObject(line, PaymentInfo.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<PaymentInfo>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<PaymentInfo>() {
                                    @Override
                                    public long extractTimestamp(PaymentInfo element, long recordTimestamp) {
                                        try {
                                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                            return sdf.parse(element.getCreate_time()).getTime();
                                        } catch (ParseException e) {
                                            e.printStackTrace();
                                            return recordTimestamp;
                                        }
                                    }
                                })
                );

//        paymentInfoDS.print("paymentInfoDS======>");

        // 3.双流Join
        SingleOutputStreamOperator<PaymentWide> paymentWideDS = paymentInfoDS.keyBy(PaymentInfo::getOrder_id)
                .intervalJoin(orderWideDS.keyBy(OrderWide::getOrder_id))
                .between(Time.minutes(-15), Time.seconds(5))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>.Context ctx, Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(paymentInfo, orderWide));
                    }
                });

        // 4.将数据写入到Kafka
        // select * from payment_info p join order_detail o on p.order_id = o.order_id;
        paymentWideDS.print("paymentWideDS======>");
        paymentWideDS.map(JSONObject::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        // 5.启动任务
        env.execute(PaymentWideApp.class.getName());

    }
}
